package com.xtayfjpk.zookeeper.test.master_workers;

import java.io.IOException;
import java.util.List;
import java.util.Random;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

import com.xtayfjpk.zookeeper.test.master_workers.callback.master.AssignTaskCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.CreateParrentCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.GetDataReassignCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.MasterCheckCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.MasterCreateCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.MasterExistsCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.MasterTasksGetChildrenCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.RecreateTaskCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.TaskDataCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.TaskDeleteCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.TaskDeletionCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.WorkerAssignmentCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.master.WorkersGetChildrenCallback;
import com.xtayfjpk.zookeeper.test.master_workers.recovery.RecoveredAssignments;
import com.xtayfjpk.zookeeper.test.master_workers.recovery.RecoveryCallback;
import com.xtayfjpk.zookeeper.test.master_workers.watcher.MasterExistsWatcher;
import com.xtayfjpk.zookeeper.test.master_workers.watcher.TasksChangeWatcher;
import com.xtayfjpk.zookeeper.test.master_workers.watcher.WorkersChangeWatcher;

public class Master implements Watcher {
	private static final Logger LOGGER = Logger.getLogger(Master.class);
	private static final String DEFAULT_HOST_PORT = "192.168.0.202:2181";
	private Random random = new Random(this.hashCode());
	private ZooKeeper zk;
	private String hostPort;
	private String serverId = Integer.toHexString(new Random().nextInt());
	boolean isLeader = false;
	private volatile boolean connected = false;
	private volatile boolean expired = false;
	
	public static enum MasterStates {RUNNING, ELECTED, NOTELECTED};

    private volatile MasterStates state = MasterStates.RUNNING;
    
    /** 缓存在子节点发生变化之前有哪些子节点 **/
	private ChildrenCache tasksCache;
	/** 缓存在子节点发生变化之前有哪些子节点 **/
	private ChildrenCache workersCache;
	
	
	public Master(String hostPort) {
		this.hostPort = hostPort;
	}
	
	public void startZK() throws IOException {
		zk = new ZooKeeper(hostPort, 15000, this);
	}
	
	public void stopZK() throws InterruptedException {
		zk.close();
	}
	
	@Override
	public void process(WatchedEvent e) {
		LOGGER.info("Processing event: " + e.toString());
		if(e.getType() == EventType.None) {
			switch(e.getState()) {
				case SyncConnected :
					connected = true;
					break;
				case Disconnected:
					connected = false;
				case Expired:
					expired = true;
					connected = false;
					LOGGER.error("Session expiration");
				default:break;
			}
		}
	}
	
    
    public void runForMaster() {
    	LOGGER.info("Running for master");
        zk.create("/master", 
                serverId.getBytes(), 
                Ids.OPEN_ACL_UNSAFE, 
                CreateMode.EPHEMERAL,
                new MasterCreateCallback(this),
                null);
    	
    }
    
	public void takeLeadership() {
		LOGGER.info("Going for list of workers");
		monitorWorkers();
		
		RecoveredAssignments assignments = new RecoveredAssignments(zk);
		assignments.setMaster(this);
		assignments.recover(new RecoveryCallback() {
			@Override
			public void recoveryComplete(int rc, List<String> tasks) {
				if(rc==RecoveryCallback.FAILED) {
					LOGGER.error("Recovery of assigned tasks failed.");
				} else {
					LOGGER.info( "Assigning recovered tasks" );
					monitorTasks();
				}
				
			}

		});
	}
	
	/** 监听workers **/
	public void monitorWorkers() {
		zk.getChildren(
				"/workers", 
				new WorkersChangeWatcher(this), 
				new WorkersGetChildrenCallback(this), 
				null);
	}
	
	/** 监听tasks，等待新的任务添加进来 **/
	public void monitorTasks() {
		zk.getChildren(
				"/tasks", 
				new TasksChangeWatcher(this), 
				new MasterTasksGetChildrenCallback(this), 
				null);		
	}
	
    
	public void reassignAndSet(List<String> children) {
		List<String> toProcess;
		if(workersCache==null) {
			workersCache = new ChildrenCache(children);
			toProcess = null;
		} else {
			LOGGER.info("Removing and setting");
			toProcess = workersCache.removedAndSet(children);
			
			//TODO 如果有新的worker加入进来呢？
		}
		
		if(toProcess!=null) {
			for(String worker : toProcess) {
				getAbsentWorkerTasks(worker);
			}
		}
	}
	
	/** 获取失联节点的任务 **/
	public void getAbsentWorkerTasks(String worker) {
		zk.getChildren(
				"/assign/"+worker, 
				false, 
				new WorkerAssignmentCallback(this), 
				null);
	}
	
	
	public void getDataReassign(String path, String task) {
		zk.getData(
				path, 
				false, 
				new GetDataReassignCallback(this), 
				task);
	}
	
	/** 重新分配任务，其实就是往/tasks节点添加子节点(待执行任务) **/
	public void recreateTask(RecreateTaskCtx ctx) {
		zk.create(
				"/tasks/" + ctx.getTask(), 
				ctx.getData(), 
				Ids.OPEN_ACL_UNSAFE, 
				CreateMode.PERSISTENT, 
				new RecreateTaskCallback(this), 
				ctx);
	}
	
	
	public void deleteAssignment(String path) {
		zk.delete(
				path, 
				-1, 
				new TaskDeletionCallback(this), 
				null);
	}
	
	
    public void checkMaster() {
    	zk.getData(
    			"/master", 
    			false, 
    			new MasterCheckCallback(this),
    			null);
    }
    
    public void exists() {
    	zk.exists(
    			"/master", 
    			new MasterExistsWatcher(this), 
    			new MasterExistsCallback(this), 
    			null);
    }
    
    public void getTaskData(String task) {
    	zk.getData("/tasks/" + task, false, new TaskDataCallback(this), task);
    }
    
    //将待执行任务分配给指定worker
    public void createAssignment(String path, byte[] data) {
    	zk.create(
    			path, 
    			data, 
    			Ids.OPEN_ACL_UNSAFE, 
    			CreateMode.PERSISTENT, 
    			new AssignTaskCallback(this), 
    			data);
    }
    
  //当待执行任务分配完成后，则需要将其在待执行任务列表中移除
    public void deleteTask(String name) {
		zk.delete(
				"/tasks/" + name, 
				-1, 
				new TaskDeleteCallback(this), 
				null);		
	}
    
    public void bootstrap() throws KeeperException, InterruptedException {
    	createParent("/workers", new byte[0]);
    	createParent("/assign", new byte[0]);
    	createParent("/tasks", new byte[0]);
    	createParent("/status", new byte[0]);
    }
    
    public void createParent(String path, byte[] data) throws KeeperException, InterruptedException {
    	zk.create(
    			path, 
    			data, 
    			Ids.OPEN_ACL_UNSAFE, 
    			CreateMode.PERSISTENT,
    			new CreateParrentCallback(this), 
    			data);
    }
    
    public MasterStates getState() {
		return state;
	}
	public void setState(MasterStates state) {
		this.state = state;
	}
	
	public String getServerId() {
		return serverId;
	}
	
	public int getWorkersSize() {
		return workersCache==null ? 0 : workersCache.getChildren().size();
	}
	
	public ChildrenCache getTasksCache() {
		return tasksCache;
	}
	public void setTasksCache(ChildrenCache tasksCache) {
		this.tasksCache = tasksCache;
	}
	public ChildrenCache getWorkersCache() {
		return workersCache;
	}
	public void setWorkersCache(ChildrenCache workersCache) {
		this.workersCache = workersCache;
	}
	public Random getRandom() {
		return random;
	}
	
	public ZooKeeper getZk() {
		return zk;
	}
	
	public static void main(String[] args) throws Exception{
		Master master = new Master(DEFAULT_HOST_PORT);
		master.startZK();
		
		while(!master.connected) {
			Thread.sleep(100);
		}
		
		master.bootstrap();
		
		master.runForMaster();
		
		while(!master.expired) {
			Thread.sleep(1000);
		}
		
		master.stopZK();
	}

}
